/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data.trans;

import cn.ac.iie.common.config.Configuration;
import cn.ac.iie.common.util.misc.MiscTools;
import cn.ac.iie.ulss.data.trans.commons.RuntimeEnv;
import cn.ac.iie.ulss.data.trans.util.Meta;
import cn.ac.iie.ulss.data.trans.worker.AllConsumer;
import cn.ac.iie.ulss.data.trans.worker.HttpSender;
import cn.ac.iie.ulss.data.trans.worker.DataProducer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import org.apache.log4j.Logger;

/**
 *
 * @author Li Mingyang
 */
public class UlssDataTrans {

    static Logger logger = Logger.getLogger(UlssDataTrans.class.getName());
    static final String CONFIGURATIONFILENAME = "data_trans.properties";

    public static void main(String[] args) {
        logger.info("intializing UlssDataTrans client...");
        try {
            init();
            run();
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }

    private static void init() throws Exception {
        String configurationFileName = "data_trans.properties";
        logger.info("getting configuration from configuration file " + configurationFileName);
        Configuration conf = Configuration.getConfiguration(configurationFileName);
        if (conf == null) {
            throw new Exception("reading " + configurationFileName + " is failed");
        }
        logger.info("initializng runtime enviroment...");
        if (!RuntimeEnv.initialize(conf)) {
            throw new Exception("initializng runtime enviroment is failed");
        }
        logger.info("initialize runtime enviroment successfully");
    }

    private static void run() throws MQClientException, IOException, Exception {
        //启动所有的发送
        initAllProducer();
        //启动所有的消费
        initAllConsumer();
        //准备停止程序
        doShutDownWork();
    }

    private static void initAllProducer() throws Exception {
        Map<String, Meta> transMetaMap = (Map<String, Meta>) RuntimeEnv.getParam(RuntimeEnv.TRANSMETAMAP);
        Set<String> keys = transMetaMap.keySet();
        for (String key : keys) {
            Meta meta = transMetaMap.get(key);
            String type = meta.getServicetype();
            if (type.equalsIgnoreCase("mq")) {
                initProducer(meta.getDataName(), meta);
            } else if (type.equalsIgnoreCase("http")) {
                initHttpSender(meta.getDataName(), meta);
            } else {
                throw new Exception("This type " + type + " is not supported.");
            }
        }
    }

    private static void initHttpSender(String dataName, Meta meta) {
        for (int i = 0; i < meta.getThreadNum(); i++) {
            Thread thread = new HttpSender(meta);
            thread.setName(dataName + " - http sender - " + i);
            thread.start();
        }
    }

    private static void initProducer(String dataName, Meta meta) {
        DefaultMQProducer producer = new DefaultMQProducer(meta.getGroupName());
        producer.setNamesrvAddr(meta.getDestinationParms().split("[-]")[0]);
        producer.setInstanceName(meta.getGroupName() + meta.getDataName() + MiscTools.getIpAddress());
        producer.setClientIP(MiscTools.getIpAddress());
        producer.setRetryTimesWhenSendFailed(3);
        producer.setSendMsgTimeout(5000);
        producer.setCompressMsgBodyOverHowmuch(32 * 1024);
        producer.setMaxMessageSize(4096 * 1024 * 2);
        producer.setClientCallbackExecutorThreads(50);
        producer.setRetryAnotherBrokerWhenNotStoreOK(true);
        try {
            producer.start();
        } catch (MQClientException ex) {
            logger.error(ex, ex);
        }
        RuntimeEnv.producers.add(producer);
        for (int i = 0; i < meta.getThreadNum(); i++) {
            Thread thread = new DataProducer(meta, producer);
            thread.setName(dataName + " - mq producer - " + i);
            thread.start();
        }
    }

    private static void initAllConsumer() {
        Thread thread = new AllConsumer();
        thread.setName("mq consumer");
        thread.start();
    }

    private static void doShutDownWork() {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                logger.info("prepare to stop trans");
                for (DefaultMQPushConsumer consumer : RuntimeEnv.consumers) {
                    consumer.shutdown();
                }
                logger.info("consumer are all down");
                Set<String> keys = RuntimeEnv.TOPIC_TO_DATAPOOL.keySet();
                for (String key : keys) {
                    while (true) {
                        if (!RuntimeEnv.TOPIC_TO_DATAPOOL.get(key).isEmpty()) {
                            try {
                                Thread.sleep(2000);
                            } catch (Exception ex) {
                                //donothing
                            }
                            logger.info("the dataPool for " + key + " is not empty! Size is " + RuntimeEnv.TOPIC_TO_DATAPOOL.get(key).size());
                            continue;
                        } else {
                            break;
                        }
                    }
                    logger.info("the dataPool for " + key + " is empty!");
                }
                for (DefaultMQProducer producer : RuntimeEnv.producers) {
                    producer.shutdown();
                }
                logger.info("producer are all down");
                logger.info("trans exit successfully");
            }
        });
    }

}
